package com.atguigu.gmall.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.text.SimpleDateFormat;

/**
 * Created on 2021/11/2.
 *
 * @author ${AUTHOR}
 */
public class BaseLagApp {

    public static final String topic = "ods_base_log";
    public static final String groupId = "atguigu";

    public static void main(String[] args) throws Exception {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:9820/gmall/flink/checkpoint"));
        System.setProperty("HADOOP_USER_NAME", "atguigu");


        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getKafkaSource(topic, groupId));
        SingleOutputStreamOperator<JSONObject> jsonObjectDS = kafkaDS.map(new MapFunction<String, JSONObject>() {
            @Override
            public JSONObject map(String value) throws Exception {

                JSONObject jsonObject = JSON.parseObject(value);
                return jsonObject;
            }
        });
        KeyedStream<JSONObject, String> keyedSD = jsonObjectDS.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject jsonObj) throws Exception {
                String mid = jsonObj.getJSONObject("common").getString("mid");

                return mid;
            }
        });

        SingleOutputStreamOperator<JSONObject> jsonObjWithIsNewDS = keyedSD.map(new RichMapFunction<JSONObject, JSONObject>() {

            private ValueState<String> firstVisitDataState;
            private SimpleDateFormat sdf;

            @Override
            public void open(Configuration parameters) throws Exception {
                firstVisitDataState = getRuntimeContext().getState(
                        new ValueStateDescriptor<String>("firstVisitDataState", String.class)
                );
                sdf = new SimpleDateFormat("yyyyMMdd");
            }

            @Override
            public JSONObject map(JSONObject jsonObj) throws Exception {

                // System.out.println(jsonObj);
                String is_new = jsonObj.getJSONObject("common").getString("is_new");
                Long ts = jsonObj.getLong("ts");
                String curVisitDate = sdf.format(ts);
                if ("1".equals(is_new)) {
                    if (firstVisitDataState.value() != null && !firstVisitDataState.value().equals("")) {
                        if (!firstVisitDataState.equals(curVisitDate)) {
                            jsonObj.getJSONObject("common").put("is_new", "0");
                        }
                    } else {
                        firstVisitDataState.update(curVisitDate);
                    }
                }
                return jsonObj;
            }
        });
        OutputTag<String> startDS = new OutputTag<String>("startTag") {
        };

        OutputTag<String> displayDS = new OutputTag<String>("displayTag") {
        };


        SingleOutputStreamOperator<String> pageDS = jsonObjWithIsNewDS.process(
                new ProcessFunction<JSONObject, String>() {
                    @Override
                    public void processElement(JSONObject jsonObject, Context context, Collector<String> collector) throws Exception {
                        JSONObject startJsonObj = jsonObject.getJSONObject("start");
                        if (startJsonObj != null && startJsonObj.size() > 0) {
                            context.output(startDS, startJsonObj.toString());
                        } else {
                            collector.collect(jsonObject.toString());
                            JSONArray displaysArr = jsonObject.getJSONArray("displays");
                            if (displaysArr != null && displaysArr.size() > 0) {
                                String pageId = jsonObject.getJSONObject("page").getString("page_id");
                                Long ts = jsonObject.getLong("ts");
                                for (int i = 0; i < displaysArr.size(); i++) {
                                    JSONObject displayJsonObj = displaysArr.getJSONObject(i);
                                    displayJsonObj.put("page_id", pageId);
                                    displayJsonObj.put("ts", ts);
                                    context.output(displayDS, displayJsonObj.toJSONString());

                                }
                            }
                        }
                    }
                }
        );
        DataStream<String> startDSOutput = pageDS.getSideOutput(startDS);
        DataStream<String> displayDSOutput = pageDS.getSideOutput(displayDS);
        startDSOutput.print("startDSOutput");
        displayDSOutput.print("displayDSOutput");
        pageDS.print("pageDS");


        // jsonObjWithIsNewDS.print("jsonObjWithIsNewDS");

        pageDS.addSink(MyKafkaUtil.getKafkaSink("dwd_start_log"));
        displayDSOutput.addSink(MyKafkaUtil.getKafkaSink("dwd_display_log"));
        startDSOutput.addSink(MyKafkaUtil.getKafkaSink("dwd_page_log"));

        env.execute();
    }
}
